ROS2 话题通信节点开发与调试

关联：索引

1. 场景提问：你写了发布者，终端显示节点在跑，为什么订阅者就是收不到？你需要哪些证据证明“话题真的在传数据”？

• 包名：小写下划线，语义清晰，例如 sorting_topic_demo
• 节点名：小写下划线，例如 material_type_pub、material_type_sub
• 话题名：建议带业务前缀，例如 /sorting/material_type
• 消息类型：本课统一使用 std_msgs/msg/String（便于快速验证）
• 参数化：频率、话题名建议做成参数（进阶要求）
• 日志：订阅回调必须打印收到的内容（作为“运行证据”）

2. 构建与依赖配置要点（学生容易踩坑）

• 依赖要同时满足两处：package.xml 与 setup.py/入口点
• 新增可执行节点后必须重新 colcon build 并重新 source install/setup.bash
• 运行失败先看：包名是否能被 ros2 pkg list 找到，入口点是否存在，Python 文件是否可导入

1. 步骤 0：创建工作空间与包（示例）

source /opt/ros/humble/setup.bash
# 每组建议设置不同 ROS_DOMAIN_ID（例如组号），避免互相串话题
export ROS_DOMAIN_ID=0

mkdir -p ~/ros2_ws/src
cd ~/ros2_ws/src
ros2 pkg create sorting_topic_demo --build-type ament_python --dependencies rclpy std_msgs

2. 步骤 1：编写发布者与订阅者源码（示例结构）

建议文件路径：

• ~/ros2_ws/src/sorting_topic_demo/sorting_topic_demo/material_type_pub.py
• ~/ros2_ws/src/sorting_topic_demo/sorting_topic_demo/material_type_sub.py

发布者（String）：

import rclpy
from rclpy.node import Node
from std_msgs.msg import String

class MeterialTypePublisher(Node):
    def __init__(self):
        super().__init__('material_type_pub')
        self.publisher_ = self.create_publisher(String, '/sorting/material_type', 10)
        timer_period = 2  # seconds
        self.timer = self.create_timer(timer_period, self.timer_callback)
        self.material_types = ['塑料', '金属', '纸张', '玻璃']
        self.index = 0

    def timer_callback(self):
        msg = String()
        msg.data = self.material_types[self.index]
        self.publisher_.publish(msg)
        self.get_logger().info(f'Publishing: {msg.data}')
        self.index = (self.index + 1) % len(self.material_types)
        
def main(args=None):
    rclpy.init(args=args)
    material_type_publisher = MeterialTypePublisher()
    rclpy.spin(material_type_publisher)
    material_type_publisher.destroy_node()
    rclpy.shutdown()
    
if __name__ == '__main__':
    main()  

订阅者（打印收到的内容）：

import rclpy
from rclpy.node import Node
from std_msgs.msg import String 

class MetrialTypeSubscriber(Node):
    def __init__(self):
        super().__init__('material_type_sub')
        self.subscription = self.create_subscription(
            String,
            '/sorting/material_type',
            self.listener_callback,
            10)
        self.subscription  # prevent unused variable warning

    def listener_callback(self, msg):
        self.get_logger().info(f'Received: {msg.data}')
        
def main(args=None):
    rclpy.init(args=args)
    material_type_subscriber = MetrialTypeSubscriber()
    rclpy.spin(material_type_subscriber)
    material_type_subscriber.destroy_node()
    rclpy.shutdown()    
    
if __name__ == '__main__':    
    main()

3. 步骤 2：配置入口点（setup.py）

在 sorting_topic_demo/setup.py 的 entry_points 中添加：

entry_points={
    'console_scripts': [
        'material_type_pub = sorting_topic_demo.material_type_pub:main',
        'material_type_sub = sorting_topic_demo.material_type_sub:main',
    ],
},

cd ~/ros2_ws
colcon build --symlink-install
source /opt/ros/humble/setup.bash
source install/setup.bash

ros2 run sorting_topic_demo material_type_pub

新开终端：

cd ~/ros2_ws
source /opt/ros/humble/setup.bash
source install/setup.bash
ros2 run sorting_topic_demo material_type_sub

• ros2 node list 能看到两个节点
• ros2 topic list 能看到 /sorting/material_type
• ros2 topic echo /sorting/material_type 能看到持续输出（订阅节点日志也算证据）

1. 快问快答：什么情况下你会“看到节点在跑但收不到数据”？验证话题通信最关键的 2 条命令是什么？

1. 话题/节点检查（先确认“对象存在”）

ros2 node list
ros2 topic list
ros2 node info /material_type_pub
ros2 topic info /sorting/material_type

2. 查看数据是否真的在流动（再确认“有数据”）

ros2 topic echo /sorting/material_type
ros2 topic hz /sorting/material_type # 用来测量这个话题的消息发布频率（帧率 / Hz）

3. 端点与 QoS 证据（定位“为什么连不上”）

ros2 topic info -v /sorting/material_type # 查看话题的相信信息

判读要点：

• 是否同时存在 Publisher 与 Subscriber 端点
• 两端消息类型是否一致
• QoS 的 Reliability/Durability 是否兼容（不兼容会导致端点无法匹配）

4. 发布测试消息（用于隔离问题）

ros2 topic pub --once /sorting/material_type std_msgs/msg/String "{data: 'test'}"

用途：手动向 /sorting/material_type 话题发布 1 条字符串消息 test

• 订阅者收得到 test：说明订阅者基本正确，问题可能在你的发布者逻辑
• 订阅者收不到 test：说明订阅者、话题名/类型或 QoS 有问题

rqt_graph

• 图中能看到发布节点、订阅节点与话题 /sorting/material_type
• 发布/订阅关系连线清晰（建议勾选 Nodes only/Topics only 简化视图）

1. 现象 A：发布无响应/订阅收不到数据

高频原因：

• 话题名不一致（多了/少了前缀、组号不同）
• 消息类型不一致（String vs 其他类型）
• 没有重新 source（编译后没 source install/setup.bash）
• 入口点错误（console_scripts 写错模块路径或 main 函数名）

证据链（按顺序做）：

1）ros2 topic list 看到目标话题

2）ros2 topic info -v 看到两端端点且类型一致

3）ros2 topic echo 能看到数据

2. 现象 B：数据延迟/频率不稳定

高频原因：

• 定时器周期设置不合理或阻塞（回调里做了耗时操作）
• Reliable + 队列过大导致堆积（“可靠地迟到”）
• 主机负载过高或网络不稳定

证据链：

• ros2 topic hz 看实际频率与抖动
• ros2 topic info -v 对照 QoS 与队列深度（depth）

3. 现象 C：QoS 不匹配（节点都在，但端点不连接）

典型场景：

• Publisher 是 Best Effort，而 Subscriber 期望 Reliable（或反过来）
• Durability 选择不一致导致历史数据行为不符合预期

证据链：

• ros2 topic info -v /sorting/material_type 查看端点 QoS，确认不兼容项

• 两端先统一为默认 QoS（depth=10），跑通后再按需求改 QoS

• QoS 改动后必须重启节点再验证

你是 ROS2 Humble 调试专家。请根据以下证据定位问题并给出可复现修复步骤：
1）我的目标：让订阅节点收到 /sorting/material_type 的数据
2）我运行的命令：<粘贴 ros2 run / colcon build 命令>
3）我的环境：Ubuntu 22.04 + ROS2 Humble；ROS_DOMAIN_ID：<粘贴>；是否已 source /opt/ros/humble/setup.bash：<是/否>
4）证据输出：
- ros2 topic list：<粘贴>
- ros2 topic info -v /sorting/material_type：<粘贴>
- 报错全文（如有）：<粘贴>
5）我已尝试过：<粘贴>
请按“原因 → 验证方法 → 修复步骤（命令）→ 复测清单”回答。